from kafka import KafkaProducer
from kafka.errors import KafkaError
import time

def send_to_kafka(topic, message):
    producer = KafkaProducer(
        bootstrap_servers=['10.48.0.4:9092'],
        value_serializer=lambda v: v.encode('utf-8')  # 将消息编码为 UTF-8 字符串
    )

    try:
        print(f"Sending message to Kafka topic '{topic}': {message}")
        # 发送消息
        future = producer.send(topic, value=message)
        result = future.get(timeout=10)  # 阻塞等待消息确认
        print(f"Message successfully sent to Kafka. Result: {result}")
    except KafkaError as e:
        print(f"Failed to send message to Kafka: {e}")
    finally:
        producer.close()

# 示例调用
if __name__ == '__main__':
    topic = 'test_topic'
    message = "Hello, Kafka!"
    send_to_kafka(topic, message)
